--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit d2b2ef54e89dd0df36d89f84c9d7732ca65796d7
Parents : 3fbe2e9
Author : Mark Qvist <mark@unsigned.io>
Date : 2024-03-19T11:03:12+01:00
Added outbound transfer progress monitoring
Changes
2 files changed, 25 insertions(+), 10 deletions(-)
Diff
diff --git a/LXMF/LXMRouter.py b/LXMF/LXMRouter.py
index f249b0c..c89c4c8 100644
--- a/LXMF/LXMRouter.py
+++ b/LXMF/LXMRouter.py
@@ -909,10 +909,6 @@ class LXMRouter:
return True
else:
return False
-
-
- ### Message Routing & Delivery ########################
- #######################################################
def handle_outbound(self, lxmessage):
lxmessage.state = LXMessage.OUTBOUND
@@ -927,6 +923,17 @@ class LXMRouter:
self.pending_outbound.append(lxmessage)
self.process_outbound()
+ def get_outbound_progress(self, lxm_hash):
+ for lxm in self.pending_outbound:
+ if lxm.hash == lxm_hash:
+ return lxm.progress
+
+ return None
+
+
+ ### Message Routing & Delivery ########################
+ #######################################################
+
def lxmf_delivery(self, lxmf_data, destination_type = None, phy_stats = None):
try:
message = LXMessage.unpack_from_bytes(lxmf_data)
@@ -1340,6 +1347,8 @@ class LXMRouter:
# to deliver the message
direct_link = self.direct_links[delivery_destination_hash]
if direct_link.status == RNS.Link.ACTIVE:
+ if lxmessage.progress == None or lxmessage.progress < 0.05:
+ lxmessage.progress = 0.05
if lxmessage.state != LXMessage.SENDING:
RNS.log("Starting transfer of "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
lxmessage.set_delivery_destination(direct_link)
@@ -1366,8 +1375,7 @@ class LXMRouter:
self.direct_links.pop(delivery_destination_hash)
lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT
else:
- # Simply wait for the link to become
- # active or close
+ # Simply wait for the link to become active or close
RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" is pending, waiting for link to become active", RNS.LOG_DEBUG)
else:
# No link exists, so we'll try to establish one, but
@@ -1383,10 +1391,12 @@ class LXMRouter:
delivery_link = RNS.Link(lxmessage.get_destination())
delivery_link.set_link_established_callback(self.process_outbound)
self.direct_links[delivery_destination_hash] = delivery_link
+ lxmessage.progress = 0.03
else:
RNS.log("No path known for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+". Requesting path...", RNS.LOG_DEBUG)
RNS.Transport.request_path(lxmessage.get_destination().hash)
lxmessage.next_delivery_attempt = time.time() + LXMRouter.PATH_REQUEST_WAIT
+ lxmessage.progress = 0.01
else:
RNS.log("Max delivery attempts reached for direct "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG)
self.fail_message(lxmessage)
diff --git a/LXMF/LXMessage.py b/LXMF/LXMessage.py
index 3cf8122..5b7721c 100644
--- a/LXMF/LXMessage.py
+++ b/LXMF/LXMessage.py
@@ -118,7 +118,7 @@ class LXMessage:
self.signature = None
self.hash = None
self.packed = None
- self.progress = None
+ self.progress = 0.0
self.state = LXMessage.DRAFT
self.method = LXMessage.UNKNOWN
self.rssi = None
@@ -324,12 +324,14 @@ class LXMessage:
if receipt:
receipt.set_delivery_callback(self.__mark_delivered)
receipt.set_timeout_callback(self.__link_packet_timed_out)
+ self.progress = 0.50
else:
if self.__delivery_destination:
self.__delivery_destination.teardown()
elif self.representation == LXMessage.RESOURCE:
self.resource_representation = self.__as_resource()
+ self.progress = 0.10
elif self.method == LXMessage.PROPAGATED:
self.state = LXMessage.SENDING
@@ -339,11 +341,13 @@ class LXMessage:
if receipt:
receipt.set_delivery_callback(self.__mark_propagated)
receipt.set_timeout_callback(self.__link_packet_timed_out)
+ self.progress = 0.50
else:
self.__delivery_destination.teardown()
elif self.representation == LXMessage.RESOURCE:
self.resource_representation = self.__as_resource()
+ self.progress = 0.10
def determine_transport_encryption(self):
@@ -387,6 +391,7 @@ class LXMessage:
def __mark_delivered(self, receipt = None):
RNS.log("Received delivery notification for "+str(self), RNS.LOG_DEBUG)
self.state = LXMessage.DELIVERED
+ self.progress = 1.0
if self.__delivery_callback != None and callable(self.__delivery_callback):
try:
@@ -397,6 +402,7 @@ class LXMessage:
def __mark_propagated(self, receipt = None):
RNS.log("Received propagation success notification for "+str(self), RNS.LOG_DEBUG)
self.state = LXMessage.SENT
+ self.progress = 1.0
if self.__delivery_callback != None and callable(self.__delivery_callback):
try:
@@ -407,6 +413,7 @@ class LXMessage:
def __mark_paper_generated(self, receipt = None):
RNS.log("Paper message generation succeeded for "+str(self), RNS.LOG_DEBUG)
self.state = LXMessage.PAPER
+ self.progress = 1.0
if self.__delivery_callback != None and callable(self.__delivery_callback):
try:
@@ -436,7 +443,7 @@ class LXMessage:
def __update_transfer_progress(self, resource):
- self.progress = resource.get_progress()
+ self.progress = 0.10 + (resource.get_progress()*0.90)
def __as_packet(self):
if not self.packed:
@@ -465,8 +472,6 @@ class LXMessage:
if not self.__delivery_destination.status == RNS.Link.ACTIVE:
raise ConnectionError("Tried to synthesize resource for LXMF message on a link that was not active")
- self.progress = 0.0
-
if self.method == LXMessage.DIRECT:
return RNS.Resource(self.packed, self.__delivery_destination, callback = self.__resource_concluded, progress_callback = self.__update_transfer_progress)
elif self.method == LXMessage.PROPAGATED:
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────